/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月15日 上午11:17:52
*/
package cn.com.guage.flink.transformation;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * @version:1.0.1
 * @Description: （对类进行功能描述）
 * @author: yangdechao
 * @date: datedate 2021年11月15日 上午11:17:52
 */
public class CoMapFunctionTest {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		DataStreamSource<Integer> integerDataStreamSource = env.fromElements(1, 2, 3);
		DataStreamSource<String> stringDataStreamSource = env.fromElements("Tom", "Jack", "Marry");
	
		// 分别执行map操作
		SingleOutputStreamOperator<Integer> IntDS = integerDataStreamSource.map(new MapFunction<Integer, Integer>() {
			/**
			* 
			*/
			private static final long serialVersionUID = 8110092523160965085L;

			public Integer map(Integer value) throws Exception {
				return value * value;
			}
		});
		SingleOutputStreamOperator<String> StrDS = stringDataStreamSource.map(x -> "hello: " + x);
		ConnectedStreams<Integer, String> connectedStreams = IntDS.connect(StrDS);
		
		/**
		 * 对于ConnectedStreams，我们需要重写CoMapFunction或CoFlatMapFunction。
		 * 这两个接口都提供了三个泛型，这三个泛型分别对应第一个输入流的数据类型、
		 * 第二个输入流的数据类型和输出流的数据类型。在重写函数时，对于CoMapFunction，
		 * map1处理第一个流的数据，map2处理第二个流的数据；对于CoFlatMapFunction，
		 * flatMap1处理第一个流的数据，flatMap2处理第二个流的数据。Flink并不能保证两个函数调用顺序，
		 * 两个函数的调用依赖于两个数据流数据的流入先后顺序，即第一个数据流有数据到达时，
		 * map1或flatMap1会被调用，第二个数据流有数据到达时，map2或flatMap2会被调用。
		 * 下面的代码对一个整数流和一个字符串流进行了connect操作。
		 */
		connectedStreams.map(new CoMapFunction<Integer, String, String>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 42100604735098915L;

			@Override
			public String map1(Integer value) throws Exception {
				return "输出结果为---------------->" + value;
			}

			@Override
			public String map2(String value) throws Exception {
				return "输出结果为---------------->" + value;
			}
		}).print();
		env.execute();
	}
}
